package cn.doitedu.demo5;

import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.googlecode.aviator.AviatorEvaluator;
import com.googlecode.aviator.Expression;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;
import java.util.HashMap;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/10
 * @Desc: 学大数据，上多易教育
 *
 *
{
"static_profile_conditions":[
{
"tag_name":"age",
"compare_type":">",
"compare_value":20
},
{
"tag_name":"value_level",
"compare_type":"=",
"compare_value":"C"
}
],
"trigger_condition":{
"event_id":"R",
"property_name":["p0","p1","p2"],
"calc_expression":"p0>p1?p0:p2",
"target_min_value":300
}

}
 *
 *
 **/
public class RuleModelCalculator_04 implements RuleCalculator{

    private String ruleId;
    JSONObject paramObject;
    Roaring64Bitmap preSelectBitmap;

    JSONObject message;

    @Override
    public String ruleId() {
        return ruleId;
    }

    @Override
    public void init(RuntimeContext runtimeContext, String ruleParamJson, Roaring64Bitmap preSelectBitmap) throws IOException {
        this.preSelectBitmap = preSelectBitmap;

        this.paramObject = JSON.parseObject(ruleParamJson);
        this.ruleId = paramObject.getString("rule_id");

        this.message = new JSONObject();
        message.put("rule_id",ruleId);


    }

    @Override
    public void calc(UserAction userAction, Collector<String> out) throws IOException {

        if(!preSelectBitmap.contains(userAction.getUser_id())) return;

        String eventId = userAction.getEvent_id();
        HashMap<String, String> properties = userAction.getProperties();

        // 取出规则参数中的触发条件
        JSONObject triggerCondition = paramObject.getJSONObject("trigger_condition");

        String targetEventId = triggerCondition.getString("event_id");
        // [p0,p1,p2]
        JSONArray propertyNames = triggerCondition.getJSONArray("property_name");
        // "p0>p1?p0:p2"
        String calcExpression = triggerCondition.getString("calc_expression");
        // 300
        int targetMinValue = triggerCondition.getIntValue("target_min_value");


        // 运算 规则参数中的表达式
        HashMap<String, Object> expressionDataMap = new HashMap<>();

        // 准备表达式所需要的变量名及变量值
        for (int i = 0; i < propertyNames.size(); i++) {
            String pName = propertyNames.getString(i);
            int pValue = Integer.parseInt(properties.get(pName));
            expressionDataMap.put(pName,pValue);
        }

        // 编译表达式
        Expression expression = AviatorEvaluator.compile(calcExpression);
        Object result = expression.execute(expressionDataMap);

        boolean ok = false;

        if(result instanceof Integer){
            int res = (int) result;
            ok = res>targetMinValue;
        }else if(result instanceof Long){
            long res = (long) result;
            ok = res>targetMinValue;
        }else if(result instanceof Double){
            double res = (double) result;
            ok = res>targetMinValue;
        }else if(result instanceof Float){
            float res = (float) result;
            ok = res>targetMinValue;
        }

        if(ok){
            message.put("user_id",userAction.getUser_id());
            message.put("hit_time",userAction.getAction_time());
            message.put("expression_value",result);
            out.collect(message.toJSONString());
        }

    }

    @Override
    public boolean isNeedPreSelectBitmap() {
        return true;
    }
}
